/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.hadoop.hdfs.qjournal.client;

import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.apache.hadoop.ipc.RemoteException;
import org.apache.hadoop.util.Time;

import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
import com.google.common.collect.Maps;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.protobuf.Message;
import com.google.protobuf.TextFormat;

/**
 * Represents a set of calls for which a quorum of results is needed.
 * @param <KEY> a key used to identify each of the outgoing calls
 * @param <RESULT> the type of the call result
 */
class QuorumCall<KEY, RESULT> {
  private final Map<KEY, RESULT> successes = Maps.newHashMap();
  private final Map<KEY, Throwable> exceptions = Maps.newHashMap();

  /**
   * Interval, in milliseconds, at which a log message will be made
   * while waiting for a quorum call.
   */
  private static final int WAIT_PROGRESS_INTERVAL_MILLIS = 1000;
  
  /**
   * Start logging messages at INFO level periodically after waiting for
   * this fraction of the configured timeout for any call.
   */
  private static final float WAIT_PROGRESS_INFO_THRESHOLD = 0.3f;
  /**
   * Start logging messages at WARN level after waiting for this
   * fraction of the configured timeout for any call.
   */
  private static final float WAIT_PROGRESS_WARN_THRESHOLD = 0.7f;

  private static final int JVM_FULLGC_STOP_THRESHOLD = 5000;

  /**
   * 将这些对Quorum的异步调用纳入到QuorumCall 对象的管理之内。
   * 如果异步调用发生了返回结果，则会调用addException 或者addResult方法来记录结果，调用者通过调用waitFor来获取结果
   */
  static <KEY, RESULT> QuorumCall<KEY, RESULT> create( Map<KEY, ? extends ListenableFuture<RESULT>> calls) {
    final QuorumCall<KEY, RESULT> qr = new QuorumCall<KEY, RESULT>();

    for (final Entry<KEY, ? extends ListenableFuture<RESULT>> e : calls.entrySet()) {
      Preconditions.checkArgument(e.getValue() != null, "null future for key: " + e.getKey());

      Futures.addCallback(e.getValue(), new FutureCallback<RESULT>() {
        @Override
        public void onFailure(Throwable t) {
          qr.addException(e.getKey(), t);
        }

        @Override
        public void onSuccess(RESULT res) {
          qr.addResult(e.getKey(), res);
        }
      });
    }
    return qr;
  }
  
  private QuorumCall() {
    // Only instantiated from factory method above
  }

  /**
   * 等待远程的quorum返回足够多的结果，即在Paxos协议认为写成功才返回
   * @param minResponses 如果响应数量大于该值，就可以立刻返回，不论这些响应是成功还是失败，否则，继续等待直到超时
   * @param minSuccesses 如果成功响应的个数大于该值，就可以立刻返回，否则继续等待直到超时
   * @param maxExceptions 如果收到的异常的个数大于该值，就立刻返回
   * @param millis 最长等待时间
   */
  public synchronized void waitFor( int minResponses, int minSuccesses, int maxExceptions, int millis, String operationName) throws InterruptedException, TimeoutException {
    // 做一个总结：刚开始如果没有拿到指定数量的响应，会等到个五六秒，后面通过这个算法，大致可能会隔一两秒，就会检查一下
    // 就会走一个while循环了，来检查一下是否有执行数量的journalnodes返回的响应，
    // 如果有指定数量的journalnodes响应，就会正常退出while循环，不管是成功还是响应，只要给响应就行了
    // 但是一旦在20s之内，没有指定数量的journalnodes给出响应，直接回抛出TimeoutException，就会导致Namenode宕机
    // 此时就可能认为 journalnodes集群整体崩溃了，比如3台journalnodes中宕机了2台，始终无法拿到指定数量的响应

    // minResponses =3, 必须要等到3个journalnodes都给回应才行，哪怕是同步失败的回应也可以
    // minSuccesses = 2, 必须至少有2个journalnodes是写入成功的
    // maxExceptions =2,最多只能有2个journalnodes是写入失败的
    // millis = 20s， 必须在20s内返回，要不然就是有2个journalnodes写入成功了，要不然就是写入失败

    long st = Time.monotonicNow(); // 进入方法的当前时间， st= 20:00:00
    // 如果等待时间达到 20s * 0.3 =6s, 开始输出日志, nextLogTime= 20:00:06
    long nextLogTime = st + (long)(millis * WAIT_PROGRESS_INFO_THRESHOLD);
    // et = 20:00:20 , 到这个时间还没得到足够的响应，就认为超时失败了
    long et = st + millis;

    StopWatch stopWatch = new StopWatch();

    // 循环等待，直到满足返回条件
    while (true) {
      // 开始计时
      stopWatch.restart();

      checkAssertionErrors();

      // 什么情况下可以退出这个while循环
      // 下面3种情况，是否拿到了足够数量的journalnodes返回的响应，不管是成功还是失败

      // 如果是在这儿出现了jvm gc的卡顿，会怎么样？
      // 卡顿了30秒之后，过来，基本上journalnodes集群没有故障的话，基本都会拿到指定数量的响应
      // 不要紧的，执行到下面直接就会成功之后返回

      // 第1种情况：写入成功的 + 写入失败的 >=3 ,3个journalnodes 不管成功或者失败，都给你返回一个响应
      if (minResponses > 0 && countResponses() >= minResponses) { return; }
      // 第2种情况：写入成功的 >=2, 只要有2个journalnodes 是写入成功的，就可以返回
      if (minSuccesses > 0 && countSuccesses() >= minSuccesses) { return; }
      // 第3种情况：写入失败的 >=2, 只要有2个journalnodes是写入失败的，也可以立马返回
      if (maxExceptions >= 0 && countExceptions() > maxExceptions) { return; }

      // 此时突然发生了namenode FullGC的异常卡顿，会导致执行QuorumCall.waitFor()方法的线程，
      // 作为一个工作线程会卡死，停在这里，不再执行。如果这里卡顿的时间过长，比如卡顿30秒，

      // 拿到了一个now当前时间， 20:00:01
      // 第2轮循环，now = 20:00:07
      // FullGC卡顿: 这里的now = 20:00:30
      long now = Time.monotonicNow();

      // 在这里出现了fullgc的卡顿，也不会出问题

      // 超过6秒，才输出日志
      // 第2轮循环，开始输出一些日志，在一定的小范围内，如果还没有等待结果，此时开始输出日志
      if (now > nextLogTime) {
        // waited = 7s, 已经等待7秒钟
        long waited = now - st;

        String msg = String.format( "Waited %s ms (timeout=%s ms) for a response for %s", waited, millis, operationName);
        if (!successes.isEmpty()) {
          msg += ". Succeeded so far: [" + Joiner.on(",").join(successes.keySet()) + "]";
        }
        if (!exceptions.isEmpty()) {
          msg += ". Exceptions so far: [" + getExceptionMapString() + "]";
        }
        if (successes.isEmpty() && exceptions.isEmpty()) {
          msg += ". No responses yet.";
        }
        if (waited > millis * WAIT_PROGRESS_WARN_THRESHOLD) {
          QuorumJournalManager.LOG.warn(msg);
        } else {
          QuorumJournalManager.LOG.info(msg);
        }
        // nextLogTime = 20:00:07 + 1s = 20:00:08
        nextLogTime = now + WAIT_PROGRESS_INTERVAL_MILLIS;
      }

      // rem = et(20:00:20) - now(20:00:01) = 19s
      // 第2轮循环： rem = et(20:00:20) - now(20:00:07) = 13s
      // FullGC卡顿: rem = 20:00:20 - 20:00:30 = -10s
      long rem = et - now;

      if (rem <= 0) {
        // 你先是拿到了当前流逝的时间，
        // 如果之前没有发生过fullgc的话，那么这个流逝的时间，其实很可能就是几微妙，不到1毫秒，
        // 如果是发生了fullgc的话，可能会导致elapsed时间是几十秒，几分钟，都有可能，
        long elapsed = stopWatch.elapsed();
        // 如果流逝掉的时间大于了JVM FullGC卡顿的时间（5秒），基本上就是出现了jvm fullgc卡顿，
        if(elapsed >= JVM_FULLGC_STOP_THRESHOLD) {
          // 此时的时间应该是20:00:30，因为jvm fullgc卡顿了30秒钟，
          // 超时时间，et：20:00:20
          et = et + elapsed; // 将et超时截止时间给延长，延期处理，et = 20:00:50
        } else {
          throw new TimeoutException();
        }
      }

//      // 代表，比如在指定的时间范围内，比如20s,没有成功的将 edits log写入到 大部分的 journal nodes 中，此时就会触发这段代码的执行，抛出异常
//      if (rem <= 0) {
//        // FullGC卡顿: 发生超时异常，导致namenode宕机
//        // 设计不合理：FullGC卡顿导致的时间流逝，是不应该计算进去的，本来journalnodes集群都返回响应了，journalnodes集群并没有宕机
//        throw new TimeoutException();
//      }

      // rem = Math.min(19,6-1=5) = 5
      // 第2轮循环： rem = Math.min(13,1) = 1
      rem = Math.min(rem, nextLogTime - now);
      // rem = Math.max(5,1) = 5
      // 第2轮循环：rem = Math.max(1,1) = 1
      rem = Math.max(rem, 1);

      // java并发里的一个API，让当前线程释放锁，阻塞等待5秒钟
      // 第二轮循环，等待1秒钟

      // 你的本意是等待几秒钟，如果在这个地方的前后，发生了jvm fullgc,会怎么样？
      wait(rem);
      // 结果在这里wait等待了几秒钟，接着jvm fullgc 又卡顿了几十秒，实际的时间变成了 20:01:20, 结果 et=20:00:50

      long elapsed = stopWatch.elapsed();
      if(elapsed - rem >= JVM_FULLGC_STOP_THRESHOLD){
        et = et + (elapsed - rem);
      }
    }
  }

  /**
   * 时间流逝的计时器，秒表，掐表
   * @author sangjiacun
   */
  class StopWatch {
    boolean isStarted; // 计时器是否打开
    long startNanos; // 计时器打开的时间（单位是纳秒）
    long elapseNanos; // 目前为止流逝掉的时间（单位是纳秒）

    public StopWatch(){

    }

    // 计时器复位
    public StopWatch reset(){
      this.isStarted = false;
      this.elapseNanos = 0;
      return this;
    }

    // 开启计时
    public StopWatch start(){
      this.isStarted = true;
      this.elapseNanos = System.nanoTime();
      return this;
    }

    // 从新开始计时
    public StopWatch restart(){
      this.reset().start();
      return this;
    }

    // 停止计时
    public StopWatch stop(){
      this.isStarted = false;
      this.elapseNanos += System.nanoTime() - this.startNanos;
      return this;
    }

    // 获取当前已经流逝掉的时间（单位是毫秒）
    public long elapsed(){
      long resultElapseNanos = 0L;
      if(isStarted){
        resultElapseNanos = System.nanoTime() - this.startNanos + elapseNanos;
      }else{
        resultElapseNanos = elapseNanos;
      }

      return TimeUnit.MILLISECONDS.convert(resultElapseNanos, TimeUnit.MILLISECONDS);
    }

    // 第一种用法：restart（复位后开始计时），stop（累加流逝时间），start（重新开始计时），stop（累加流逝时间）
    // 第二种用法：restart（复位后开始计时），elapsedMs（获取当前的流逝时间），来进行一定的判断，是否发生了JVM FullGC
    // 后面如果要开始重新计时的话，再次restart（复位后开始计时）
  }

  /**
   * Check if any of the responses came back with an AssertionError.
   * If so, it re-throws it, even if there was a quorum of responses.
   * This code only runs if assertions are enabled for this class,
   * otherwise it should JIT itself away.
   * 
   * This is done since AssertionError indicates programmer confusion
   * rather than some kind of expected issue, and thus in the context
   * of test cases we'd like to actually fail the test case instead of
   * continuing through.
   */
  private synchronized void checkAssertionErrors() {
    boolean assertsEnabled = false;
    assert assertsEnabled = true; // sets to true if enabled
    if (assertsEnabled) {
      for (Throwable t : exceptions.values()) {
        if (t instanceof AssertionError) {
          throw (AssertionError)t;
        } else if (t instanceof RemoteException &&
            ((RemoteException)t).getClassName().equals(
                AssertionError.class.getName())) {
          throw new AssertionError(t);
        }
      }
    }
  }

  private synchronized void addResult(KEY k, RESULT res) {
    successes.put(k, res);
    notifyAll();
  }
  
  private synchronized void addException(KEY k, Throwable t) {
    exceptions.put(k, t);
    notifyAll();
  }
  
  /**
   * @return the total number of calls for which a response has been received,
   * regardless of whether it threw an exception or returned a successful
   * result.
   */
  public synchronized int countResponses() {
    return successes.size() + exceptions.size();
  }
  
  /**
   * @return the number of calls for which a non-exception response has been
   * received.
   */
  public synchronized int countSuccesses() {
    return successes.size();
  }
  
  /**
   * @return the number of calls for which an exception response has been
   * received.
   */
  public synchronized int countExceptions() {
    return exceptions.size();
  }

  /**
   * @return the map of successful responses. A copy is made such that this
   * map will not be further mutated, even if further results arrive for the
   * quorum.
   */
  public synchronized Map<KEY, RESULT> getResults() {
    return Maps.newHashMap(successes);
  }

  public synchronized void rethrowException(String msg) throws QuorumException {
    Preconditions.checkState(!exceptions.isEmpty());
    throw QuorumException.create(msg, successes, exceptions);
  }

  public static <K> String mapToString(
      Map<K, ? extends Message> map) {
    StringBuilder sb = new StringBuilder();
    boolean first = true;
    for (Map.Entry<K, ? extends Message> e : map.entrySet()) {
      if (!first) {
        sb.append("\n");
      }
      first = false;
      sb.append(e.getKey()).append(": ")
        .append(TextFormat.shortDebugString(e.getValue()));
    }
    return sb.toString();
  }

  /**
   * Return a string suitable for displaying to the user, containing
   * any exceptions that have been received so far.
   */
  private String getExceptionMapString() {
    StringBuilder sb = new StringBuilder();
    boolean first = true;
    for (Map.Entry<KEY, Throwable> e : exceptions.entrySet()) {
      if (!first) {
        sb.append(", ");
      }
      first = false;
      sb.append(e.getKey()).append(": ")
        .append(e.getValue().getLocalizedMessage());
    }
    return sb.toString();
  }
}
